package com.jzl.canal_consumer.etl.et.rocketmq.consumer.impl;

import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.jzl.canal_consumer.etl.et.rocketmq.connector.CanalRocketMqConnector;
import com.jzl.canal_consumer.etl.et.rocketmq.consumer.CanalRocketMqConsumer;
import com.jzl.canal_consumer.etl.load.mysql.Loader;
import com.jzl.canal_consumer.etl.et.rocketmq.transform.MqDataTransform;
import com.jzl.util.LoggerUtil;
import org.springframework.util.Assert;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author ZhenWuWang
 */
public class DefaultCanalRocketMqConsumer<E, T> implements CanalRocketMqConsumer
{
    private volatile boolean running = false;

    private final CanalRocketMqConnector<E> connector;

    private final MqDataTransform<E, T> transform;

    private final Loader<T> loader;

    private final String schemaName;

    public DefaultCanalRocketMqConsumer(CanalRocketMqConnector<E> connector, MqDataTransform<E, T> transform, Loader<T> loader,
                                        String schemaName)
    {
        this.connector = connector;
        this.transform = transform;
        this.loader = loader;
        this.schemaName = schemaName;
    }

    @Override
    public void start()
    {
        Assert.notNull(connector, "connector is null");
        running = true;
        process();
        String msg = String.format("consumer start groupName：%s topic:%s schemaName:%s",
                connector.getGroupName(), connector.getTopic(), schemaName);
        LoggerUtil.info(msg);
    }

    private List<E> getMqMessage()
    {
        return connector.getListWithoutAck(1000L, TimeUnit.MILLISECONDS);
    }

    /**
     * 数据转换
     *
     * @param message 从rocketMq拿出来的消息
     * @return 转换后的数据
     */
    protected List<T> transform(List<E> message)
    {
        return transform.transform(schemaName, message);
    }

    /**
     * 加载数据
     *
     * @param list 需要加载的数据
     */
    private void load(List<T> list)
    {
        loader.load(list);
    }

    @Override
    public void stop()
    {
        if (!running)
        {
            return;
        }
        running = false;
    }

    private void process()
    {
        while (running)
        {
            try
            {
                connector.connect();
                connector.subscribe();
                while (running)
                {
                    //拿数据
                    List<E> message = getMqMessage();
                    if (message != null && message.size() > 0)
                    {
                        //转换数据
                        List<T> transformList = transform(message);
                        //加载数据
                        if (transformList != null && transformList.size() > 0)
                        {
                            load(transformList);
                        }
                    }
                    connector.ack(); // 提交确认
                }
            } catch (Exception e)
            {
                LoggerUtil.error(e.getMessage(), e);
            }
        }
        try
        {
            connector.unsubscribe();
        } catch (CanalClientException e)
        {
            // No-op. Continue process
            e.printStackTrace();
        }
//
//        if (running)
//        {
//            //TODO 重连
//        }
    }
}
